package com.xs.producer;

import java.io.IOException;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Title: <P>Produce.java</P>
 * @Package <P>com.xs.produce</P>
 * @Description: <P>TODO该类...</P>
 * @author renweiping
 * @date 2019年3月11日 下午12:44:16
 * @version 1.0
 */

public class Produce {
	private static final Logger logger = LoggerFactory.getLogger(Produce.class);
	private static final String topic ="test2";
	private String content;
	private static final String brokerUrl = "127.0.0.1:9092";
	private static KafkaProducer<String,String> producer = null;

	private  Properties initConfig() {
		Properties config = new Properties();
		config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		//0 生产者成功写入消息之前不会等待来自任何服务器的响应，这种配置，提高吞吐量，但是消息存在丢失风险。
		//1 只要集群的leader（master）收到了消息，生产者将会受到发送成功的一个响应，如果消息无撞到达首领节点（比如首领节点崩愤，新的首领还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。不过，如果一个没有收到消息的节点成为新首领，消息还是会丢失。这个时候的吞吐量取决于使用的是
		//同步发送还是异步发送。如果让发送客户端等待服务器的响应（通过调用Futu re 对象的get（）方法，显然会增加延迟（在网络上传输一个来回的延迟）。如果客户端使用回调，延迟问题就可以得到缓解，不过吞吐量还是会受发送中消息数量的限制（比如，生产者在收到服务器响应之前可以发送多少个消息）
		//ALL 所有参与复制的节点全部收到消息的时候，生产者才会收到来自服务器的一个响应，这种模式最安全，但是吞吐量受限制，它可以保证不止一个服务器收到消息，就算某台服务器奔溃，那么整个集群还是会正产运转。
		config.put(ProducerConfig.ACKS_CONFIG, "1");
		////重新发送消息次数，到达次数返回错误
		config.put(ProducerConfig.RETRIES_CONFIG, "0");
		////Producer会尝试去把发往同一个Partition的多个Requests进行合并，batch.size指明了一次Batch合并后Requests总大小的上限。如果这个值设置的太小，可能会导致所有的Request都不进行Batch。
		config.put("batch.size", 163840);
		////Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms则更进一步，这个参数为每次发送增加一些delay，以此来聚合更多的Message。
		config.put("linger.ms", 1);
		//在Producer端用来存放尚未发送出去的Message的缓冲区大小
		config.put("buffer.memory", 33554432);
		config.put("request.timeout.ms", "60000");
		return config;
	}
	

	public Produce(String content) {
		this.content = content ;
		Properties config = initConfig();
		producer = new KafkaProducer<String,String>(config);
	
	}
	
	public void sendMsg() throws IOException{
		ProducerRecord<String, String> record = null;
		for(int i =0 ;i<10;i++){
			String content = this.content+"_"+i;
			record = new ProducerRecord<String,String>(topic,content);
			producer.send(record);
			logger.info("KafkaProducer.send(\"{}\", {}) OK.",
					topic, content);
			//producer.flush();
		}
		//producer.close();
		//System.in.read();
	}
	
	public void sendMsgSync(){
		ProducerRecord<String, String> record = null;
		for(int i =0 ;i<100;i++){
			final String key = i+"";
			final String value = this.content+"_"+i;
			record = new ProducerRecord<String,String>(topic,key,value);
			producer.send(record,new Callback() {
				
				@Override
				public void onCompletion(RecordMetadata meta, Exception e) {
					if(e!=null){
						e.printStackTrace();
					}else{
						logger.info("KafkaProducer.push(\"{}\", {}, {}, {}, \"{}\") OK.",
	                            meta.topic(), meta.partition(), meta.offset(), key, value);
					}
					
				}
			});
			producer.flush();
			
		}
	}

	public void sendMsg(ProducerRecord<String,String> record, Callback call){
		producer.send(record, call);
		producer.flush();
	}
	
	
	
}
